/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.office;

import org.apache.mesos.MesosNativeLibrary;
import org.apache.mesos.Protos.*;

import java.util.Collection;


/**
 * Concrete implementation of a SchedulerDriver that connects a
 * Scheduler with a Mesos master. The MesosSchedulerDriver is
 * thread-safe.
 * <p/>
 * Note that scheduler failover is supported in Mesos. After a
 * scheduler is registered with Mesos it may failover (to a new
 * process on the same machine or across multiple machines) by
 * creating a new driver with the ID given to it in {@link
 * Scheduler#registered}.
 * <p/>
 * The driver is responsible for invoking the Scheduler callbacks as
 * it communicates with the Mesos master.
 * <p/>
 * Note that blocking on the MesosSchedulerDriver (e.g., via {@link
 * #join}) doesn't affect the scheduler callbacks in anyway because
 * they are handled by a different thread.
 * <p/>
 * <p/>
 * Note that the driver uses GLOG to do its own logging. GLOG flags can
 * be set via environment variables, prefixing the flag name with
 * "GLOG_", e.g., "GLOG_v=1". For Mesos specific logging flags see
 * src/logging/flags.hpp. Mesos flags can also be set via environment
 * variables, prefixing the flag name with "MESOS_", e.g.,
 * "MESOS_QUIET=1".
 * <p/>
 * See src/examples/java/TestFramework.java for an example of using
 * the MesosSchedulerDriver.
 */
public class MesosSchedulerDriver implements SchedulerDriver
{
    static
    {
        MesosNativeLibrary.load();
    }

    /**
     * Creates a new driver for the specified scheduler. The master
     * should be one of:
     * <pre>
     * {@code
     *     host:port
     *     zk://host1:port1,host2:port2,.../path
     *     zk://username:password@host1:port1,host2:port2,.../path
     *     file:///path/to/file (where file contains one of the above)
     * }
     * </pre>
     * <p/>
     * The driver will attempt to "failover" if the specified
     * FrameworkInfo includes a valid FrameworkID.
     * <p/>
     * Any Mesos configuration options are read from environment
     * variables, as well as any configuration files found through the
     * environment variables.
     * <p/>
     *
     * @param scheduler The scheduler implementation which callbacks are invoked
     *                  upon scheduler events.
     * @param framework The frameworkInfo describing the current framework.
     * @param master    The address to the currently active Mesos master.
     */
    // TODO(vinod): Deprecate this in favor the constructor that takes
    //              'credential' as parameter.
    public MesosSchedulerDriver(Scheduler scheduler, FrameworkInfo framework, String master)
    {
        if (scheduler == null)
        {
            throw new NullPointerException("Not expecting a null Scheduler");
        }

        if (framework == null)
        {
            throw new NullPointerException("Not expecting a null FrameworkInfo");
        }

        if (master == null)
        {
            throw new NullPointerException("Not expecting a null master");
        }

        this.scheduler = scheduler;
        this.framework = framework;
        this.master = master;
        this.implicitAcknowledgements = true;
        this.credential = null;

        initialize();
    }

    /**
     * Same as the other constructors, except that it accepts the newly
     * introduced 'credential' parameter.
     *
     * @param scheduler  The scheduler implementation which callbacks are invoked
     *                   upon scheduler events.
     * @param framework  The frameworkInfo describing the current framework.
     * @param master     The address to the currently active Mesos master.
     * @param credential The credentials that will be used used to authenticate
     *                   calls from this scheduler.
     */
    public MesosSchedulerDriver(Scheduler scheduler, FrameworkInfo framework, String master, Credential credential)
    {

        if (scheduler == null)
        {
            throw new NullPointerException("Not expecting a null Scheduler");
        }

        if (framework == null)
        {
            throw new NullPointerException("Not expecting a null FrameworkInfo");
        }

        if (master == null)
        {
            throw new NullPointerException("Not expecting a null master");
        }

        if (credential == null)
        {
            throw new NullPointerException("Not expecting a null credential");
        }

        this.scheduler = scheduler;
        this.framework = framework;
        this.master = master;
        this.implicitAcknowledgements = true;
        this.credential = credential;

        initialize();
    }

    /**
     * Same as the other constructors, except that it accepts the newly
     * introduced 'implicitAcknowledgements' parameter.
     *
     * @param scheduler                The scheduler implementation which callbacks are invoked
     *                                 upon scheduler events.
     * @param framework                The frameworkInfo describing the current framework.
     * @param master                   The address to the currently active Mesos master.
     * @param implicitAcknowledgements Whether the driver should send
     *                                 acknowledgements on behalf of the scheduler. Setting this to
     *                                 false allows schedulers to perform their own acknowledgements,
     *                                 which enables asynchronous / batch processing of status updates.
     */
    public MesosSchedulerDriver(Scheduler scheduler,
                                FrameworkInfo framework,
                                String master,
                                boolean implicitAcknowledgements)
    {

        if (scheduler == null)
        {
            throw new NullPointerException("Not expecting a null Scheduler");
        }

        if (framework == null)
        {
            throw new NullPointerException("Not expecting a null FrameworkInfo");
        }

        if (master == null)
        {
            throw new NullPointerException("Not expecting a null master");
        }

        this.scheduler = scheduler;
        this.framework = framework;
        this.master = master;
        this.implicitAcknowledgements = implicitAcknowledgements;
        this.credential = null;

        initialize();
    }

    /**
     * Same as the other constructors, except that it accepts the newly
     * introduced 'implicitAcknowledgements' and 'credentials' parameters.
     *
     * @param scheduler                The scheduler implementation which callbacks are invoked
     *                                 upon scheduler events.
     * @param framework                The frameworkInfo describing the current framework.
     * @param master                   The address to the currently active Mesos master.
     * @param implicitAcknowledgements Whether the driver should send
     *                                 acknowledgements on behalf of the scheduler. Setting this to
     *                                 false allows schedulers to perform their own acknowledgements,
     *                                 which enables asynchronous / batch processing of status updates.
     * @param credential               The credentials that will be used used to authenticate
     *                                 calls from this scheduler.
     */
    public MesosSchedulerDriver(Scheduler scheduler,
                                FrameworkInfo framework,
                                String master,
                                boolean implicitAcknowledgements,
                                Credential credential)
    {

        if (scheduler == null)
        {
            throw new NullPointerException("Not expecting a null Scheduler");
        }

        if (framework == null)
        {
            throw new NullPointerException("Not expecting a null FrameworkInfo");
        }

        if (master == null)
        {
            throw new NullPointerException("Not expecting a null master");
        }

        if (credential == null)
        {
            throw new NullPointerException("Not expecting a null credential");
        }

        this.scheduler = scheduler;
        this.framework = framework;
        this.master = master;
        this.implicitAcknowledgements = implicitAcknowledgements;
        this.credential = credential;

        initialize();
    }

    public native Status start();

    public native Status stop(boolean failover);

    public Status stop()
    {
        return stop(false);
    }

    public native Status abort();

    public native Status join();

    public Status run()
    {
        Status status = start();
        return status != Status.DRIVER_RUNNING ? status : join();
    }

    public native Status requestResources(Collection<Request> requests);

    public Status launchTasks(OfferID offerId,
                              Collection<TaskInfo> tasks)
    {
        return launchTasks(offerId, tasks, Filters.newBuilder().build());
    }

    public native Status launchTasks(OfferID offerId,
                                     Collection<TaskInfo> tasks,
                                     Filters filters);

    public Status launchTasks(Collection<OfferID> offerIds,
                              Collection<TaskInfo> tasks)
    {
        return launchTasks(offerIds, tasks, Filters.newBuilder().build());
    }

    public native Status launchTasks(Collection<OfferID> offerIds,
                                     Collection<TaskInfo> tasks,
                                     Filters filters);

    public native Status killTask(TaskID taskId);

    public native Status acceptOffers(Collection<OfferID> offerIds,
                                      Collection<Offer.Operation> operations,
                                      Filters filters);

    public Status declineOffer(OfferID offerId)
    {
        return declineOffer(offerId, Filters.newBuilder().build());
    }

    public native Status declineOffer(OfferID offerId, Filters filters);

    public native Status reviveOffers();

    public native Status suppressOffers();

    public native Status acknowledgeStatusUpdate(TaskStatus status);

    public native Status sendFrameworkMessage(ExecutorID executorId,
                                              SlaveID slaveId,
                                              byte[] data);

    public native Status reconcileTasks(Collection<TaskStatus> statuses);

    protected native void initialize();

    protected native void finalize();

    private final Scheduler scheduler;
    private final FrameworkInfo framework;
    private final String master;
    private final boolean implicitAcknowledgements;
    private final Credential credential;

    private long __scheduler;
    private long __driver;
}
